package com.dyyx.nettyrpc.server;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dyyx.nettyrpc.protocol.JsonUtil;
import com.dyyx.nettyrpc.protocol.RpcRequest;
import com.dyyx.nettyrpc.protocol.RpcResponse;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import net.sf.cglib.reflect.FastClass;


public class RpcHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger logger = LoggerFactory.getLogger(RpcHandler.class);

    private final Map<String, Object> handlerMap;

    public RpcHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, final RpcRequest request) throws Exception {
        RpcServer.submit(new Runnable() {
            @Override
            public void run() {
            	if(logger.isDebugEnabled()){
                    logger.debug("Receive request:" + JsonUtil.objectToJson(request));
            	}
                RpcResponse response = new RpcResponse();
                response.setRequestId(request.getRequestId());
                try {
                    Object result = handle(request);
                    
                    
                    if(logger.isDebugEnabled()){
                        logger.debug("request[{}].result={}",request.getRequestId(),result);
                	}
                    
                    response.setResult(result);
                } catch (Throwable t) {
                    response.setError(t.toString());
                    logger.error("RPC Server handle request error", t);
                }
                ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    	if(logger.isDebugEnabled()){
                            logger.debug("Send response for request " + request.getRequestId());
                    	}
                    }
                });
            }
        });
    }

    private Object handle(RpcRequest request) throws Throwable {
        String className   = request.getClassName();
        Object serviceBean = handlerMap.get(className);

        Class<?>   serviceClass   = serviceBean.getClass();
        String     methodName     = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[]   parameters     = request.getParameters();

      
		// JDK reflect
		// Method method = serviceClass.getMethod(methodName, parameterTypes);
		// method.setAccessible(true);
		// return method.invoke(serviceBean, parameters);

		// Cglib reflect
		FastClass serviceFastClass = FastClass.create(serviceClass);
		// FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName,
		// parameterTypes);
		// return serviceFastMethod.invoke(serviceBean, parameters);
		// for higher-performance
		int methodIndex = serviceFastClass.getIndex(methodName, parameterTypes);
		return serviceFastClass.invoke(methodIndex, serviceBean, parameters);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("RpcHandler error", cause);
        ctx.close();
    }
}
